package com.shuhe.thread;

import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.JsonNode;
import com.shuhe.common.BottomLevelTableGenerator;
import com.shuhe.common.MyDateUtils;
import com.shuhe.domain.alarm.TableSchema;
import com.shuhe.domain.alarm.TopicTable;
import com.shuhe.metastore.DBAppSql;
import org.apache.log4j.Logger;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.*;

import static com.shuhe.common.CommonFunction.getUUID;

public class MqttReceiveThread implements Runnable {
    private ArrayList<String> execs;

    private static Logger logger = Logger.getLogger(MqttReceiveThread.class);

    private String brokerUrl;
    private String client_id;
    private String username;
    private String password;
    private ArrayList<TopicTable> topicTables;
    private Connection appConn;

    private String start_time = MyDateUtils.getNowDate();
    private String end_time = MyDateUtils.getNowDate();

    private MqttClient client = null;

    public MqttReceiveThread(String brokerUrl, String client_id, String username, String password,
                             ArrayList<TopicTable> topicTables, Connection appConn) {
        this.brokerUrl = brokerUrl;
        this.client_id = client_id;
        this.username = username;
        this.password = password;
        this.topicTables = topicTables;
        this.appConn = appConn;
    }

    public String jsonToSql(JsonNode firstElement, String tableName, String topic) {
        String sql = "INSERT INTO " + tableName + " (";
        String valueStr = "";
        Iterator<Map.Entry<String, JsonNode>> fields = firstElement.fields();
        while (fields.hasNext()) {
            Map.Entry<String, JsonNode> field = fields.next();
            String fieldName = field.getKey();
            JsonNode value = field.getValue();
            sql += fieldName + ", ";
            valueStr += "'" + value.asText() + "', ";
        }
        sql += "ods_uuid, ods_timestamp, topic) VALUES (";
        valueStr = valueStr + "'" + getUUID() + "','" + MyDateUtils.getNowDate() + "', '" + topic + "')";
        return sql + valueStr;
    }

    // 重连后重新订阅主题
    private void resubscribeToTopics(MqttClient client) {
        try {
            for (int i = 0; i < topicTables.size(); i++) {
                client.subscribe(topicTables.get(i).getTopic(), Integer.parseInt("0"));
            }
        } catch (MqttException e) {
            System.err.println("重新订阅失败: " + e.getMessage());
        }
    }

    public void run() {
        try {
            client = new MqttClient(brokerUrl, client_id, new MemoryPersistence());
            MqttConnectionOptions options = new MqttConnectionOptions();
            options.setUserName(username);
            options.setPassword(password.getBytes());
            options.setCleanStart(true);
            options.setAutomaticReconnect(true); // 启用自动重连
            options.setKeepAliveInterval(60); // 设置心跳间隔（秒）
            options.setSessionExpiryInterval(0L);

            // 设置回调
            client.setCallback(new MqttCallback() {
                @Override
                public void disconnected(MqttDisconnectResponse disconnectResponse) {
                    logger.info("与 Broker 连接断开: " + disconnectResponse.getReasonString() + ": " + brokerUrl + ":" + client_id);
                }

                @Override
                public void mqttErrorOccurred(MqttException e) {
                    String errorMessage = "MQTT 错误: " + brokerUrl + ":" + client_id + e.getMessage();
                    writeErrorLog(errorMessage, "");
                    logger.error(errorMessage);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    int pos = -1;
                    String dest_table_name = "";
                    boolean is_exist = false;
                    for (int i = 0; i < topicTables.size(); i++) {
                        if (topicTables.get(i).getTopic().equals(topic)) {
                            dest_table_name = topicTables.get(i).getDest_table_name();
                            is_exist = topicTables.get(i).isIs_exist();
                            pos = i;
                            break;
                        }
                    }
                    try {
                        String messageStr = new String(message.getPayload());
                        messageStr = JSONObject.parse(messageStr).toString();
                        List<TableSchema> schemas = BottomLevelTableGenerator.generateBottomLevelTables(messageStr,
                                dest_table_name);
                        if (!is_exist) {
                            // 创建表
                            DBAppSql.execSql(appConn, schemas.get(0).toSQL(dest_table_name));
                            topicTables.get(pos).setIs_exist(true);
                        }
                        // 插入数据
                        JsonNode jsonNode = schemas.get(0).getJsonNode();
                        if (jsonNode.isArray()) {
                            for (int i = 0; i < jsonNode.size(); i++) {
                                JsonNode jsonNode1 = jsonNode.get(i);
                                String sql = jsonToSql(jsonNode1, dest_table_name, topic);
                                DBAppSql.execSql(appConn, sql);
                            }
                        } else {
                            String sql = jsonToSql(jsonNode, dest_table_name, topic);
                            DBAppSql.execSql(appConn, sql);
                        }
                        if (topicTables.get(pos).isIs_blank()) {
                            //插入样例数据
                            String sql = "update sys_alarminfo_collection_config_subscribe_topic set data_sample = '"
                                    + messageStr + "' where topic = '" + topic + "'";
                            DBAppSql.execSql(appConn, sql);
                            topicTables.get(pos).setIs_blank(false);
                        }

                    } catch (Exception e) {
                        String errorMessage = "MQTT 错误: " + ": " + brokerUrl + ":" + client_id + e.getMessage();
                        writeErrorLog(errorMessage, topicTables.get(pos).getSubscribe_id());
                        logger.error(errorMessage);
                    }
                }

                @Override
                public void deliveryComplete(IMqttToken token) {
                    // 主要用于发布消息的确认，订阅方可不重点实现
                }

                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    logger.info("连接成功: " + (reconnect ? "重连" : "首次连接") + ": " + brokerUrl + ":" + client_id);
                    if (reconnect) {
                        // 重要：重连后重新订阅
                        resubscribeToTopics(client);
                    }
                }

                @Override
                public void authPacketArrived(int reasonCode, MqttProperties properties) {
                    // 处理认证包，仅在需要特殊认证时使用
                }
            });

            // 连接到 Broker
            client.connect(options);

            // 订阅主题
            for (int i = 0; i < topicTables.size(); i++) {
                client.subscribe(topicTables.get(i).getTopic(), Integer.parseInt("0"));
            }
            // 保持程序运行以持续接收消息
            int i = 0;
            while (true) {
                Thread.sleep(1000 * 60);
                end_time = MyDateUtils.getBeforeMinuteBydate(start_time, 1);
                if (client.isConnected()){
                    writeSuccessLog();
                } else{
                    writeErrorLog("MQTT 错误: " + brokerUrl + ":" + client_id + ": " + "连接断开", "");
                }

                i++;
                if (i > 60 * 12) {
                    client.disconnect();
                    client.close();
                    break;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            String errorMessage = "MQTT 错误: " + brokerUrl + ":" + client_id + e.getMessage();
            writeErrorLog(errorMessage, "");
            logger.error(errorMessage);
        } finally {
            if (client != null) {
                try {
                    client.disconnect();
                    client.close();
                } catch (MqttException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private void writeSuccessLog() {
        try {
            for (int i = 0; i < topicTables.size(); i++) {
                TopicTable topicTable = topicTables.get(i);
                if (topicTable.isIs_exist()) {
                    String sql = "select count(1) from " + topicTable.getDest_table_name() + " where ods_timestamp >= "
                            + "'" + start_time + "'" + " and ods_timestamp < " + "'" + end_time + "'";
                    ResultSet rs = DBAppSql.querySql(appConn, sql);
                    if (rs.next()) {
                        int count = rs.getInt(1);
                        sql = "insert into sys_alarminfo_collection_log(collection_id, subscribe_id, start_time, " +
                                "end_time, data_count, message) values (" + "'" + getUUID() + "','"
                                + topicTable.getSubscribe_id() + "','" + start_time + "','" + end_time + "','"
                                + count + "', '执行成功')";
                        DBAppSql.execSql(appConn, sql);
                    }
                }
            }
            start_time = end_time;
        } catch (Exception e) {
            System.err.println("写入日志失败: " + e.getMessage());
        }
    }

    private void writeErrorLog(String errorMessage, String subscribe_id) {
        try {
            //写采集日志
            errorMessage = errorMessage.replace("'", "''");
            String sql = "insert into sys_alarminfo_collection_log(collection_id, subscribe_id, start_time, " +
                    "end_time, data_count, message) values (" + "'" + getUUID() + "','"
                    + subscribe_id + "', now(), now(), '', '执行失败：" + errorMessage + "')";
            DBAppSql.execSql(appConn, sql);

            writeErrorMessage(appConn, errorMessage, "msg_type_01");
        } catch (Exception e) {
            System.err.println("写入日志失败: " + e.getMessage());
        }
    }

    public static void writeErrorMessage(Connection appConn, String errorMessage, String msg_type_code) {
        try {
            //写消息日志
            errorMessage = errorMessage.replace("'", "''");
            String sql = "insert into sys_message_detail(id, msg_type_code, msg_detail, msg_time, status) values ("
                    + "'" + getUUID() + "','" + msg_type_code + "', '" + errorMessage + "', now(), '0')";
            DBAppSql.execSql(appConn, sql);

        } catch (Exception e) {
            System.err.println("写入日志失败: " + e.getMessage());
        }
    }
}
